/*
 * Copyright (C) 2017 Pingan, Inc. All Rights Reserved.
 */

package com.pingan.haofang.searchcloud.es.search;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.reflect.TypeToken;
import com.pingan.haofang.searchcloud.api.IndexRowData;
import com.pingan.haofang.searchcloud.api.matedata.FieldMeta;
import com.pingan.haofang.searchcloud.api.matedata.constant.MetadataFieldType;
import com.pingan.haofang.searchcloud.api.matedata.constant.SQLConstant;
import com.pingan.haofang.searchcloud.api.query.parser.dbobject.Condition;
import com.pingan.haofang.searchcloud.api.query.parser.dbobject.ConditionGroup;
import com.pingan.haofang.searchcloud.api.query.parser.dbobject.SqlNode;
import com.pingan.haofang.searchcloud.api.query.parser.dbobject.sqlnode.ColumnNode;
import com.pingan.haofang.searchcloud.api.query.parser.dbobject.sqlnode.function.RangeGroup;
import com.pingan.haofang.searchcloud.api.query.parser.dbobject.sqlnode.function.RangeGroupGapNode;
import com.pingan.haofang.searchcloud.api.query.parser.element.QueryBlock;
import com.pingan.haofang.searchcloud.api.query.parser.element.SelectStatement;
import com.pingan.haofang.searchcloud.api.route.AliasUtils;
import com.pingan.haofang.searchcloud.es.api.ElasticSearchSearchService;
import com.pingan.haofang.searchcloud.es.api.SearchResultDTO;
import com.pingan.haofang.searchcloud.es.api.client.EsClientFactory;
import com.pingan.haofang.searchcloud.es.api.client.RestClientFactory;
import com.pingan.haofang.searchcloud.es.dto.ElkLogResultDTO;
import com.pingan.haofang.searchcloud.es.dto.WrapResultDTO;
import com.pingan.haofang.searchcloud.es.dto.WrapSourceDTO;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.expression.BinaryExpression;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.ExpressionVisitor;
import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
import net.sf.jsqlparser.expression.operators.relational.LikeExpression;
import net.sf.jsqlparser.expression.operators.relational.OldOracleJoinBinaryExpression;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.statement.select.SelectVisitor;
import net.sf.jsqlparser.util.deparser.ExpressionDeParser;
import net.sf.jsqlparser.util.deparser.SelectDeParser;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.plugin.nlpcn.QueryActionElasticExecutor;
import org.elasticsearch.search.SearchHits;
import org.nlpcn.es4sql.SearchDao;
import org.nlpcn.es4sql.jdbc.ObjectResult;
import org.nlpcn.es4sql.jdbc.ObjectResultsExtractor;
import org.nlpcn.es4sql.query.QueryAction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Es查询接口
 *
 * @author zhangbi617
 * @date 2017-07-31
 */
@Component
public class ElasticSearchSqlSearchService implements ElasticSearchSearchService {

    private static Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();

    @Autowired
    private EsClientFactory esClientFactory;

    @Autowired
    RestClientFactory restClientFactory;

    private ObjectResult getObjectResult(Object execution,
                                         boolean flat,
                                         boolean includeScore,
                                         boolean includeType,
                                         boolean includeId) throws Exception {

        return (new ObjectResultsExtractor(includeScore, includeType, includeId)).extractResults(execution, flat);
    }

    /**
     * 查询替换为别名进行
     *
     * @param sql
     * @return
     * @throws JSQLParserException
     */
    private String replaceTableName(String sql) throws JSQLParserException {
        Statement statement = CCJSqlParserUtil.parse(sql);
        Select select = (Select) statement;

        StringBuilder buffer = new StringBuilder();
        ExpressionDeParser expressionDeParser = new ExpressionDeParser();
        TableNameParser tableNameParser = new TableNameParser(expressionDeParser, buffer);

        expressionDeParser.setSelectVisitor(tableNameParser);
        expressionDeParser.setBuffer(buffer);
        select.getSelectBody().accept(tableNameParser);

        return select.toString();
    }

    public class TableNameParser extends SelectDeParser {

        public TableNameParser(ExpressionVisitor expressionVisitor, StringBuilder buffer) {
            super(expressionVisitor, buffer);
        }

        @Override
        public void visit(Table tableName) {
            tableName.setName(AliasUtils.convertAliasName(tableName.getName()));
        }
    }


    private String parseCondition(String sql, Map<String, ?> map, Class<? extends ExpressionDeParser> clazz) throws Exception {
        Statement statement = CCJSqlParserUtil.parse(sql);
        Select select = (Select) statement;

        StringBuilder buffer = new StringBuilder();

        SelectDeParser selectDeParser = new SelectDeParser();
        Constructor<?>[] constructors = clazz.getConstructors();
        ExpressionDeParser parser = (ExpressionDeParser) constructors[0].newInstance(this,selectDeParser, buffer, map);

        selectDeParser.setExpressionVisitor(parser);
        selectDeParser.setBuffer(buffer);

        select.getSelectBody().accept(selectDeParser);

        return buffer.toString();
    }

    /**
     * 替换距离
     *
     * @param sql
     * @param statement
     * @return
     * @throws JSQLParserException
     */
    private String replaceGeo(String sql, SelectStatement statement) throws Exception {
        //获得所有geo类型
        List<String> geoList = Lists.newArrayList();
        Map<String, FieldMeta> fieldMetaMap = statement.getMeta().getFiledMetas();
        for (Map.Entry<String, FieldMeta> entry : fieldMetaMap.entrySet()) {
            if (entry.getValue().getType() == MetadataFieldType.GEO) {
                geoList.add(entry.getKey());
            }
        }
        Map<String, GeoQueryDTO> geoQueryMap = new HashMap<>();

        //获得所有GEO
        QueryBlock queryBlock = statement.getQueryBlock();
        if (CollectionUtils.isNotEmpty(queryBlock.getWhereConditionGroupList())) {
            buildQueryCondition(queryBlock.getWhereConditionGroupList(), geoQueryMap, geoList);
        }

        //经纬度调换
        geoQueryMap.entrySet().stream().map(Map.Entry::getValue).map(this::convertGeo)
                .collect(Collectors.toMap(GeoQueryDTO::getName, Function.identity(), (o, n) -> (n)));

        return parseCondition(sql, geoQueryMap, GeoConditionParser.class);
    }


    public class GeoConditionParser extends ExpressionDeParser {

        public GeoConditionParser(SelectVisitor selectVisitor, StringBuilder buffer, Map<String, GeoQueryDTO> geoQueryMap) {
            super(selectVisitor, buffer);
            this.geoQueryMap = geoQueryMap;
        }

        private Map<String, GeoQueryDTO> geoQueryMap;

        public void visitOldOracleJoinBinaryExpression(OldOracleJoinBinaryExpression expression, String operator) {
            if (expression.isNot()) {
                this.getBuffer().append(" NOT ");
            }

            if (SQLConstant.GEO_D_KEY.equals(expression.getLeftExpression().toString())) {
                this.getBuffer().append("1=1");
                return;
            }

            if (geoQueryMap.keySet().contains(expression.getLeftExpression().toString())) {
                String key = expression.getLeftExpression().toString();
                this.getBuffer().append("GEO_DISTANCE(" + key + ",'" + geoQueryMap.get(key).getDistance() + "km'," + geoQueryMap.get(key).getPoint() + ")");
                return;
            }

            if (SQLConstant.Q_KEY.equals(expression.getLeftExpression().toString())) {
                this.getBuffer().append(SQLConstant.DF_SEARCH_KEY);
            }else{
                expression.getLeftExpression().accept(this);
            }

            if (expression.getOldOracleJoinSyntax() == 1) {
                this.getBuffer().append("(+)");
            }

            this.getBuffer().append(operator);
            expression.getRightExpression().accept(this);
            if (expression.getOldOracleJoinSyntax() == 2) {
                this.getBuffer().append("(+)");
            }
        }
    }

    public class IkConditionParser extends ExpressionDeParser{

        public IkConditionParser(SelectVisitor selectVisitor, StringBuilder buffer, Map<String, Object> ikKeyMap) {
            super(selectVisitor, buffer);
            this.ikKeyMap = ikKeyMap;
        }

        private Map<String, Object> ikKeyMap;

        private void visitBinaryExpression(BinaryExpression binaryExpression, String operator) {
            if (binaryExpression.isNot()) {
                this.getBuffer().append(" NOT ");
            }

            binaryExpression.getLeftExpression().accept(this);
            this.getBuffer().append(operator);
            binaryExpression.getRightExpression().accept(this);
        }

        public void visit(LikeExpression likeExpression) {

            if(ikKeyMap.containsKey(likeExpression.getLeftExpression().toString())){
                String content =  likeExpression.getRightExpression().toString();
                this.getBuffer().append(likeExpression.getLeftExpression().toString());
                this.getBuffer().append("=");
                this.getBuffer().append("matchQuery("+content+")");
                return;
            }else{
                this.visitBinaryExpression(likeExpression, " LIKE ");
                String escape = likeExpression.getEscape();
                if (escape != null) {
                    this.getBuffer().append(" ESCAPE '").append(escape).append('\'');
                }
            }
        }
    }

    private static void buildQueryCondition(List<ConditionGroup> groupList, Map<String, GeoQueryDTO> geoQueryMap, List<String> geoList) {

        for (ConditionGroup group : groupList) {
            if (!group.isChild()) {
                buildQueryCondition(group.getChildList(), geoQueryMap, geoList);
            } else {
                if (group.getDistance() != null) {
                    for (Condition con : group.getConditionList()) {
                        if (geoList.contains(con.getLeft().getColumnName())) {
                            geoQueryMap.put(con.getLeft().getColumnName(),
                                    new GeoQueryDTO(con.getLeft().getColumnName(), String.valueOf(con.getRightOne().getValue()), group.getDistance()));
                        }
                    }
                }
            }
        }
    }

    private GeoQueryDTO convertGeo(GeoQueryDTO geoQueryDTO) {
        String[] pointArr = geoQueryDTO.getPoint().split(",");
        geoQueryDTO.setPoint(pointArr[1] + "," + pointArr[0]);
        return geoQueryDTO;
    }

    /**
     * 替换rangegroup
     *
     * @param sql
     * @param statement
     * @return
     * @throws Exception
     */
    private String replaceRangeGroup(String sql, SelectStatement statement) throws Exception {
        Map<String, LinkedList<Integer>> rangeGroupMap = new HashMap<>();
        if (statement.getQueryBlock().getGroupByClause() != null) {
            List<? extends SqlNode> nodes = statement.getQueryBlock().getGroupByClause();
            for (SqlNode node : nodes) {
                if (node instanceof RangeGroup) {
                    SqlNode column = ((RangeGroup) node).getParams().get(0);
                    String key = ((ColumnNode) column).getColumnName();
                    LinkedList<Integer> gapList = new LinkedList();
                    for (RangeGroupGapNode gapNode : ((RangeGroup) node).getGapNodeList()) {
                        Integer t = gapNode.getBegin();
                        while (t < gapNode.getEnd()) {
                            gapList.add(t);
                            t += gapNode.getGap();
                        }
                        gapList.add(gapNode.getEnd());
                    }
                    rangeGroupMap.put(key, gapList);
                }
            }
        }
        if (rangeGroupMap.size() > 0) {
            sql = parseCondition(sql, rangeGroupMap, RangeGroupParser.class);
        }
        return sql;
    }


    public class RangeGroupParser extends ExpressionDeParser {

        public RangeGroupParser(SelectVisitor selectVisitor, StringBuilder buffer, Map<String, LinkedList<Integer>> rangeGroupMap) {
            super(selectVisitor, buffer);
            this.rangeGroupMap = rangeGroupMap;
        }

        private Map<String, LinkedList<Integer>> rangeGroupMap;

        private boolean useBracketsInExprList = true;

        public void visit(ExpressionList expressionList) {
            if (this.useBracketsInExprList) {
                this.getBuffer().append("(");
            }

            Iterator iter = expressionList.getExpressions().iterator();

            while (iter.hasNext()) {
                Expression expression = (Expression) iter.next();
                if (expression instanceof Column) {
                    LinkedList<Integer> gaplist = rangeGroupMap.getOrDefault(((Column) expression).getColumnName(), new LinkedList<>());
                    if (CollectionUtils.isEmpty(gaplist)) return;
                    this.getBuffer().append(((Column) expression).getColumnName());
                    Iterator gapiter = gaplist.iterator();
                    while (gapiter.hasNext()) {
                        this.getBuffer().append(", ");
                        this.getBuffer().append(gapiter.next());
                    }
                    break;
                } else {
                    expression.accept(this);
                    if (iter.hasNext()) {
                        this.getBuffer().append(", ");
                    }
                }
            }

            if (this.useBracketsInExprList) {
                this.getBuffer().append(")");
            }
        }

        public void visit(net.sf.jsqlparser.expression.Function function) {
            if (function.isEscaped()) {
                this.getBuffer().append("{fn ");
            }
            if ("range_group".equals(function.getName())) {
                this.getBuffer().append("range");
                this.visit(function.getParameters());
                return;
            } else {
                this.getBuffer().append(function.getName());
            }

            if (function.isAllColumns() && function.getParameters() == null) {
                this.getBuffer().append("(*)");
            } else if (function.getParameters() == null) {
                this.getBuffer().append("()");
            } else {
                boolean oldUseBracketsInExprList = this.useBracketsInExprList;
                if (function.isDistinct()) {
                    this.useBracketsInExprList = false;
                    this.getBuffer().append("(DISTINCT ");
                } else if (function.isAllColumns()) {
                    this.useBracketsInExprList = false;
                    this.getBuffer().append("(ALL ");
                }

                super.visit(function.getParameters());
                this.useBracketsInExprList = oldUseBracketsInExprList;
                if (function.isDistinct() || function.isAllColumns()) {
                    this.getBuffer().append(")");
                }
            }

            if (function.getAttribute() != null) {
                this.getBuffer().append(".").append(function.getAttribute());
            }

            if (function.isEscaped()) {
                this.getBuffer().append("}");
            }
        }
    }

    @Override
    public SearchResultDTO query(String host, String sql, SelectStatement statement) throws Exception {
        SearchResultDTO dto = new SearchResultDTO();

        sql = replaceTableName(sql);

        //ik分词
        sql = replaceIk(sql, statement);
        //距离d=
        sql = replaceGeo(sql, statement);
        //range_group
        sql = replaceRangeGroup(sql, statement);

        long before = System.currentTimeMillis();

        SearchDao searchDao = new SearchDao(esClientFactory.getByHost(host));
        QueryAction queryAction = searchDao.explain(sql);
        Object execution = QueryActionElasticExecutor.executeAnyAction(searchDao.getClient(), queryAction);

        ObjectResult result = getObjectResult(execution,
                true, false, false, true);

        dto.setResultColumns(Sets.newHashSet(result.getHeaders()));

        List<IndexRowData> indexRowDatas = new ArrayList<>();
        for (List<Object> line : result.getLines()) {
            IndexRowData indexRowData = new IndexRowData();
            for (int i = 0; i < result.getHeaders().size(); i++) {
                indexRowData.build(result.getHeaders().get(i), line.get(i));
            }
            indexRowDatas.add(indexRowData);
        }

        dto.setResultSize(indexRowDatas.size());
        if (execution instanceof SearchHits) {
            dto.setTotalFound(((SearchHits) execution).getTotalHits());
        } else {
            dto.setTotalFound(indexRowDatas.size());
        }
        dto.setResult(indexRowDatas);
        dto.setTime((System.currentTimeMillis() - before) / 1000);

        return dto;
    }

    /**
     * 替换ik查询
     * @param sql
     * @param statement
     * @return
     * @throws Exception
     */
    private String replaceIk(String sql, SelectStatement statement) throws Exception {
        Map<String,Object> ikKeyMap=new HashMap();

        statement.getMeta().getFiledMetas().forEach((k,v)->{
            if(v.getType() == MetadataFieldType.MY_IK){
                ikKeyMap.put(k,v);
            }
        });

        if(ikKeyMap.size()>0){
            sql = parseCondition(sql, ikKeyMap, IkConditionParser.class);
        }
        return sql;
    }

    @Override
    public String explain(String host, String sql, SelectStatement statement) {
        try {
            sql = replaceTableName(sql);
            sql = replaceGeo(sql, statement);
            sql = replaceRangeGroup(sql, statement);
            //ik分词
            sql = replaceIk(sql, statement);
            SearchDao searchDao = new SearchDao(esClientFactory.getByHost(host));
            QueryAction queryAction = searchDao.explain(sql);
            return queryAction.explain().explain();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public SearchResultDTO query(String collection, String params, String host) throws Exception {

        RestClient restClient = restClientFactory.getByHost(host);

        String[] nameArr = collection.split("_");
        String[] expectedIndexNameArr = new String[nameArr.length - 1];
        System.arraycopy(nameArr, 0, expectedIndexNameArr, 0, expectedIndexNameArr.length);
        String configName = StringUtils.join(expectedIndexNameArr, "_");

        String resultJson = this.performSearchWithBody("POST", params,
                String.format("searchcloud_%s", collection), String.format("searchcloud_%s", configName), restClient);

        Type type = new TypeToken<ElkLogResultDTO<WrapResultDTO<WrapSourceDTO<IndexRowData>>>>() {
        }.getType();
        ElkLogResultDTO<WrapResultDTO<WrapSourceDTO<IndexRowData>>> source =
                GSON.fromJson(resultJson, type);

        List<WrapSourceDTO<IndexRowData>> wrapSourceDTOList = Optional.ofNullable(source).map(ElkLogResultDTO::getHits)
                .map(WrapResultDTO::getHits).orElse(Collections.emptyList());

        List<IndexRowData> dataList = Lists.newArrayList();
        for (WrapSourceDTO<IndexRowData> wrapDTO : wrapSourceDTOList) {
            IndexRowData data = wrapDTO.get_source();
            dataList.add(data);
        }

        SearchResultDTO searchResultDTO = new SearchResultDTO();
        searchResultDTO.setTime(source.getTook());
        searchResultDTO.setTotalFound(source.getHits().getTotal());
        searchResultDTO.setResultSize(dataList.size());
        searchResultDTO.setResult(dataList);

        return searchResultDTO;
    }

    public String performSearchWithBody(String methord, String paramJson, String index, String type, RestClient client) {
        try {
            Response response = client.performRequest(methord, StringUtils.join("/", index, "/", type, "/_search"),
                    Collections.singletonMap("pretty", "true"), new NStringEntity(paramJson, ContentType.APPLICATION_JSON));
            return EntityUtils.toString(response.getEntity(), "utf-8");
        } catch (IOException e) {
        }
        return null;
    }


}
